package com.juanpi.order.runner;

import com.juanpi.yinhe.configuration.ConfigService;
import com.juanpi.yinhe.configuration.Shoe;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author xueqing
 *         Created on 2017/2/13.
 */
public class RunnerHolder {

    @Autowired
    private PushHttpClient pushHttpClient;

    @Autowired
    private ConfigService configService;

    private ExecutorService executorService = Executors.newCachedThreadPool();

    /**
     *
     */
    ConcurrentMap<String,Cow> concurrentMap = new ConcurrentHashMap();
    public void start(){

        List<Shoe> allShoes = configService.findAllShoes();
        for(Shoe shoe:allShoes){
            Properties properties = getProperties(shoe);
            UrlTopicMappingBean urlTopicMappingBean = getUrlTopicMappingBean(shoe);
            Cow cow = new Cow(properties, urlTopicMappingBean, pushHttpClient);
            executorService.submit(cow);
        }


    }

    private Properties getProperties(Shoe shoe) {
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers",shoe.getService());
        properties.setProperty("group.id",shoe.getGroupId());
        properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("enable.auto.commit","false");
        properties.setProperty("auto.commit.interval.ms","1000000");
        properties.setProperty("fetch.min.bytes","1");
        return properties;
    }

    private UrlTopicMappingBean getUrlTopicMappingBean(Shoe shoe) {
        UrlTopicMappingBean urlTopicMappingBean = new UrlTopicMappingBean();
        urlTopicMappingBean.setHeader(shoe.getHeader());
        urlTopicMappingBean.setTopic(shoe.getTopic());
        urlTopicMappingBean.setUrl(shoe.getUrl());
        return urlTopicMappingBean;
    }
}

